package net.opentsdb.tsd;

import net.opentsdb.core.TSDB;
import net.opentsdb.stats.QueryStats;
import net.opentsdb.tools.ArgP;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.handler.codec.http.*;
import org.jboss.netty.util.CharsetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author Hujf
 * @title: AbstractHttpQuery
 * @date 2021-09-23 10:42
 * @description: 抽象队列处理查询请求
 */
public abstract class AbstractHttpQuery {
    private static final Logger LOG = LoggerFactory.getLogger(AbstractHttpQuery.class);

    /**
     * The request in this HTTP query.
     */
    private final HttpRequest request;
    /**
     * The channel on which the request was received.
     */
    private final Channel chan;
    /**
     * Shortcut to the request method
     */
    private final HttpMethod method;
    /**
     * The {@code TSDB} instance we belong to
     */
    protected final TSDB tsdb;
    /**
     * Used for recording query statistics
     */
    protected QueryStats stats;

    /**
     * When the query was started (useful for timing).
     */
    private final long start_time = System.nanoTime();

    /**
     * The response object we'll fill with data
     */
    private final DefaultHttpResponse response =
            new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
    private Map<String,List<String>> querystring;

    public HttpRequest request() {
        return request;
    }

    protected AbstractHttpQuery(final TSDB tsdb, final HttpRequest request, final Channel chan) {
        this.tsdb = tsdb;
        this.request = request;
        this.chan = chan;
        this.method = request.getMethod();
    }

    public abstract String getQueryBaseRoute();

    public void notFound() {
        LOG.warn("Not Found: " + request.getUri());
        sendStatusOnly(HttpResponseStatus.NOT_FOUND);
    }

    public void sendStatusOnly(final HttpResponseStatus status) {
        if (!chan.isConnected()) {
            if (stats != null) {
                stats.markSendFailed();
            }
            done();
            return;
        }

        response.setStatus(status);
        final boolean keepalive = HttpHeaders.isKeepAlive(request);
        if (keepalive) {
            HttpHeaders.setContentLength(response, 0);
        }
        final ChannelFuture future = chan.write(response);
//        if (stats != null) {
//            future.addListener(new SendSuccess());
//        }
        if (!keepalive) {
            future.addListener(ChannelFutureListener.CLOSE);
        }
        done();
    }

    /**
     * Returns the HTTP method/verb for the request
     */
    public HttpMethod method() {
        return this.method;
    }

    /**
     * Sends an HTTP reply to the client.
     *
     * @param status The status of the request (e.g. 200 OK or 404 Not Found).
     * @param buf    The content of the reply to send.
     */
    public void sendBuffer(final HttpResponseStatus status,
                           final ChannelBuffer buf,
                           final String contentType) {
        if (!chan.isConnected()) {
            if (stats != null) {
                stats.markSendFailed();
            }
            done();
            return;
        }
        response.headers().set(HttpHeaders.Names.CONTENT_TYPE, contentType);

        // TODO(tsuna): Server, X-Backend, etc. headers.
        // only reset the status if we have the default status, otherwise the user
        // already set it
        response.setStatus(status);
        response.setContent(buf);
        final boolean keepalive = HttpHeaders.isKeepAlive(request);
        if (keepalive) {
            HttpHeaders.setContentLength(response, buf.readableBytes());
        }
        final ChannelFuture future = chan.write(response);
        if (stats != null) {
            //   future.addListener(new SendSuccess());
        }
        if (!keepalive) {
            future.addListener(ChannelFutureListener.CLOSE);
        }
        done();
    }

    /**
     * Method to call after writing the HTTP response to the wire.  The default
     * is to simply log the request info.  Can be overridden by subclasses.
     */
    public void done() {
        final int processing_time = processingTimeMillis();
        final String url = request.getUri();
        final String msg = String.format("HTTP %s done in %d ms", url, processing_time);
        if (url.startsWith("/api/put") && LOG.isDebugEnabled()) {
            // NOTE: Suppresses too many log lines from /api/put.
            LOG.debug(msg);
        } else {
            LOG.info(msg);
        }
        LOG.info("HTTP " + request.getUri() + " done in " + processing_time + "ms");
    }

    /**
     * Returns how many ms have elapsed since this queryHttpQuery was created.
     */
    public int processingTimeMillis() {
        return (int) ((System.nanoTime() - start_time) / 1000000);
    }

    public boolean hasContent() {
        return this.request.getContent() != null &&
                this.request.getContent().readable();
    }

    public String getContext(){
        return this.request.getContent().toString(this.getCharset());
    }

    private Charset getCharset() {
        for(String type:this.request.headers().getAll("Context-Type")){
            int idx = type.toUpperCase().indexOf("CHARSET=");
            if(idx>1){
                String charset = type.substring(idx+8);
                return Charset.forName(charset);
            }
        }
        return  StandardCharsets.UTF_8;
    }

    public  boolean hasQueryStringParam(String name){
        return getQueryString().get(name)!=null;
    }

    private Map<String, List<String>> getQueryString() {
        if(querystring==null){
            querystring = new QueryStringDecoder(request.getUri()).getParameters();
        }
        return querystring;
    }

    public String getQueryStringParam(String param) {
        List<String> params = getQueryString().get(param);
        return params==null?null:params.get(params.size()-1);
    }

    public  List<String> getQueryStringParams(String param){
        return   getQueryString().get(param);
    }

    public String getRemoteAddress() {
        return chan.getRemoteAddress().toString();
    }

    public  Map<String,String> getPrintableHeaders(){
        Map <String,String>headers = new HashMap<>(request.headers().entries().size());

        for(final Map.Entry<String,String> header:request.headers().entries()){
            headers.put(header.getKey(),header.getValue());
        }
        return headers;
    }

    public void setStat(QueryStats stat) {
        this.stats = stat;
    }

    public QueryStats getStat() {
        return stats;
    }
}
